package com.chengqs.dfs.backupnode.server;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;

/**
 * <B>主类名称：</B>EditLogTrailer<BR>
 * <B>概要说明：</B>enter<BR>
 *
 * @author: chengqs
 * @create: 2025-03-31 21:18:37
 **/
@Slf4j
public class EditsLogFetcher extends Thread {

    public static final Integer BACKUP_NODE_FETCH_SIZE = 10;

    private BackupNode backupNode;
    private NameNodeRpcClient nameNode;
    private FSNameSystem nameSystem;


    public EditsLogFetcher(BackupNode backupNode, FSNameSystem nameSystem, NameNodeRpcClient nameNode) {
        this.backupNode = backupNode;
        this.nameNode = nameNode;
        this.nameSystem = nameSystem;
    }

    @Override
    public void run() {
        log.info("EditsLog抓取线程已经启动......");

        while (backupNode.isRunning()) {
            try {
                if (!nameSystem.isFinishedRecover()) {
                    log.info("当前还没完成元数据的恢复，不进行editLog同步......");
                    Thread.sleep(1000);
                    continue;
                }

                long syncedTxid = nameSystem.getSyncedTxid();
                JSONArray editsLogs = nameNode.fetchEditsLog(syncedTxid);

                if (editsLogs.size() == 0) {
                    // log.info("没有拉取到任何一条editslog，等待1秒后继续尝试拉取");
                    Thread.sleep(1000);
                    continue;
                }

                if (editsLogs.size() < BACKUP_NODE_FETCH_SIZE) {
                    Thread.sleep(1000);
                    // log.info("拉取到的edits log不足10条数据，等待1秒后再次继续拉取");
                }

                for (int i = 0; i < editsLogs.size(); i++) {
                    JSONObject editsLog = editsLogs.getJSONObject(i);
                    log.info("拉取到一条editslog：{}", editsLog);
                    String op = editsLog.getString("OP");
                    if (op.equals("MKDIR")) {
                        String path = editsLog.getString("PATH");
                        try {
                            nameSystem.mkdir(editsLog.getLongValue("txid"), path);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    } else if (op.equals("CREATE")) {
                        String filename = editsLog.getString("PATH");
                        try {
                            nameSystem.create(filename);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }

                nameNode.setNameNodeRunning(true);
            } catch (Exception e) {
                // e.printStackTrace();
                nameNode.setNameNodeRunning(false);
            }
        }
    }
}
